import { Injectable, OnModuleDestroy, OnModuleInit } from "@nestjs/common"
// import kafka from 'kafka-node'
import { Kafka, logLevel } from 'kafkajs'
import { Collection, MongoClient } from 'mongodb';
import { RedisService } from "./redis.service";
;
import dayjs from 'dayjs'

import { LBQueue } from '@utils/lb.queue'
 
@Injectable()
export class MongoService implements OnModuleInit, OnModuleDestroy {
    // client: kafka.KafkaClient = null;
    // consumer: kafka.Consumer = null;
    mongo: MongoClient = null;
    msgColl: Collection = null;

    ctx: LBQueue = new LBQueue()

    kafkaHost: string = '134.175.224.228:9092'

    constructor(
        private redisService: RedisService,
        //private msgService : MsgService
    ) { }

    async onModuleInit() {
        this.mongo = new MongoClient('mongodb://134.175.224.228:27017', {
            auth: {
                username: 'root',
                password: '88888888'
            },
        });

        const db = await this.mongo.db('gewe');

        this.msgColl = await db.collection('msg');

        const kafka = new Kafka({
            logLevel: logLevel.INFO,
            brokers: [`${this.kafkaHost}`],
            clientId: 'kafka-push',
        })

        const topic = 'MsgToMongo'
        const consumer = kafka.consumer({ groupId: 'mongo' })

        const run = async () => {
            await consumer.connect()
            await consumer.subscribe({ topic, fromBeginning: true })
            await consumer.run({
                eachMessage: async ({ topic, partition, message }) => {
                    let data: any = JSON.parse(message.value.toString())
                    if (data?.Appid) {
                        await this.pushMessageToMongo(data)
                    }
                }
            })
        }

        run().catch((e) =>
            console.log(e)
        )

        // setInterval(async () => {
        //     let ctx = await this.ensureConversationExist()

        //     带环-循环队列
        //     读取队列 
        //         await ctx.read(5, async (index) => {
        //             if (index < 0) {
        //                 console.log('队列无消息..')
        //                 return
        //             }

        //             let res = await this.msgColl.aggregate([
        //                 {
        //                     $match: { doc_id: "si_2062181717_8171668571" }
        //                 },
        //                 {
        //                     $project: {
        //                         _id: 0,
        //                         // 使用$arrayElemAt获取数组指定索引的元素
        //                         msg: { $arrayElemAt: ["$msgs", index] },
        //                     }
        //                 },
        //             ]).toArray();

        //             //let res = await this.msgColl.findOne({ doc_id: "si_2062181717_8171668571" },projection)
        //             console.log(`读索:${index} ==>${dayjs(res[0].msg.Data.MsgSeq).format('YYYY-MM-DD HH:mm:ss')}`)
        //             console.log(`最小可读索引${ctx.backIndex}:==>${dayjs(res[0].min.Data.MsgSeq).format('YYYY-MM-DD HH:mm:ss.SSS')}`)
        //             console.log(`最大可读索引${ctx.forwardReadableIndex()}:==>${dayjs(res[0].max.Data.MsgSeq).format('YYYY-MM-DD HH:mm:ss.SSS')}`)
        //         })

        //         await this.updateContext(ctx)
        // }, 5000)


        //检查 时间边界
        // setInterval(async () => {
        //     let res = await this.msgColl.aggregate([
        //         {
        //             $match: { doc_id: "si_2062181717_8171668571" }
        //         },
        //         {
        //             $project: {
        //                 _id: 0,
        //                 // 使用$arrayElemAt获取数组指定索引的元素
        //                 forwardIndex: {
        //                     $mod: [{ $subtract: ['$forwardIndex', 1] }, "$maxBufferLen"]
        //                 },
        //                 backIndex: 1,
        //                 max: {
        //                     $let: {
        //                         vars: {
        //                             indexRaw: {
        //                                 $mod: [{ $subtract: ['$forwardIndex', 1] }, "$maxBufferLen"]
        //                             }
        //                         },
        //                         in: {
        //                             $arrayElemAt: ["$msgs", "$$indexRaw"] // 获取数组的第一个元素
        //                             //$multiply : ["$$indexRaw" , 1]
        //                         }
        //                     },
        //                 },
        //                 min: {
        //                     $arrayElemAt: ["$msgs", "$backIndex"] // 获取数组的第一个元素
        //                 }
        //             }
        //         }
        //     ]).toArray()

        //     //let res = await this.msgColl.findOne({ doc_id: "si_2062181717_8171668571" },projection)
        //     console.log(`最小可读索引${res[0].backIndex}:==>${dayjs(res[0].min.Data.MsgSeq).format('YYYY-MM-DD HH:mm:ss.SSS')}`)
        //     console.log(`最大可读索引${res[0].forwardIndex}:==>${dayjs(res[0].max.Data.MsgSeq).format('YYYY-MM-DD HH:mm:ss.SSS')}`)
        // }, 4000)


        // setInterval(async () => {
        //     let res = await this.msgColl.aggregate([
        //         {
        //             $match: { doc_id: "si_2062181717_8171668571" }
        //         },
        //         {
        //             $project: {
        //                 _id: 0,
        //                 newests: {
        //                     $slice: [{
        //                         $sortArray: {
        //                             input: '$msgs',
        //                             sortBy: {
        //                                 "Data.MsgSeq": 1
        //                             }
        //                         }
        //                     }, 5]
        //                 },
        //             }
        //         }]).toArray()

        //     for (const msg of res[0].newests) {
        //         console.log(`${dayjs(msg.Data.MsgSeq).format('YYYY-MM-DD HH:mm:ss.SSS') }`)
        //     }

        // }, 3000)

        // this.client = new kafka.KafkaClient({ kafkaHost: '134.175.224.228:9092' })

        // this.consumer = new kafka.Consumer(this.client, [{ topic: 'MsgToMongo', partition: 0 }], { groupId: 'mongo', autoCommit: false })

        // this.consumer.on('error', (msg) => {
        //     console.log('error')
        // })

        // this.consumer.on('message', async (msg) => {
        //     let data: any = JSON.parse(msg.value as string)
        //     if (data?.Appid) {
        //         // if (data.Data.MsgSeq > 100)
        //         //     return

        //         await this.pushMessageToMongo(data)

        //         // 提交offset
        //         this.consumer.commit((error, data) => {
        //             // if (error) console.error(error);
        //             // else console.log('提交到MongoDB ', data);
        //         });
        //     } else {
        //         console.log(msg)
        //     }
        // })
    }

    onModuleDestroy() {

    }

    getMaxSeqKey(conversationID) {
        return `MAX_SEQ:${conversationID}`
    }

    async getMaxSeq(Wxid:any,conversationID:any) {
        let maxSeq = null //await this.redisService.redis().get(this.getMaxSeqKey(conversationID)) as any

        let doc_id = `${Wxid}_${conversationID}`;
        let ctx = await this.ensureConversationExist(doc_id);

        if (maxSeq == null) {
            let res = await this.msgColl.findOne({ doc_id},
                {
                    projection: {
                        maxSeq: { $size: "$msgs" }
                    }
                }
            );

            // await this.msgColl.updateOne({ doc_id: "si_2062181717_8171668571" },
            //     {
            //         $setOnInsert: {
            //             maxSeq: 0,
            //             msgs: []
            //         }
            //     }, { upsert: true })

            // let res = await this.msgColl.findOneAndUpdate({ doc_id: "si_2062181717_8171668571" },
            //     {
            //         $inc: {
            //             maxSeq: 1
            //         }
            //     },{
            //         projection: { maxSeq:1}
            //     })

            return res.maxSeq
        } else {
            return maxSeq
        }
    }

    async persistenceToMongo() {
        let range = {
            maxBufferLength: 100,
            forwordIndex: -1,
            backIndex: -1
        }
    }

    async ensureConversationExist(doc_id,ctx: LBQueue = null): Promise<LBQueue> {
        if (!ctx)
            ctx = new LBQueue()

        let res = await this.msgColl.findOneAndUpdate({doc_id},
            {
                $setOnInsert: {
                    maxSeq: 0,
                    msgs: [],
                    maxBufferLen: 200,
                    forwardIndex: 0,
                    backIndex: 0
                }
            },
            {
                upsert: true,
                projection: { maxBufferLen: 1, forwardIndex: 1, backIndex: 1 }, // 指定返回的字段
                returnDocument: 'after' // 指定是返回更新前的文档还是更新后的文档
            }
        )

        // res = res.toArray()
        ctx.forwardIndex = res.forwardIndex
        ctx.backIndex = res.backIndex
        ctx.maxBufferLen = res.maxBufferLen

        return ctx
    }

    async updateContext(doc_id : any,ctx: LBQueue, update: object = null) {
        if (!update)
            update = {}

        if (update['$set'] == undefined || update['$set'] == null) {
            update['$set'] = {}
        }

        update['$set'][`forwardIndex`] = ctx.forwardIndex
        update['$set'][`backIndex`] = ctx.backIndex
        update['$set'][`maxBufferLen`] = ctx.maxBufferLen

        await this.msgColl.updateOne({doc_id},
            update,
            { upsert: true }
        )

        // let res = await this.msgColl.aggregate([
        //     {
        //         $match: { doc_id: "si_2062181717_8171668571" }
        //     },
        //     {
        //         $project: {
        //             _id: 0,
        //             max: {
        //                 $arrayElemAt: ["$msgs", "$forwardIndex"] // 获取数组的第一个元素
        //             },
        //             min: {
        //                 $arrayElemAt: ["$msgs", "$backIndex"] // 获取数组的第一个元素
        //             }
        //         }
        //     },
        // ]).toArray();

        // //let res = await this.msgColl.findOne({ doc_id: "si_2062181717_8171668571" },projection)
        // console.log(`最小索引${ctx.backIndex}:==>${dayjs(res[0].min.Data.MsgSeq).format('YYYY-MM-DD HH:mm:ss')}`)
        // console.log(`最新索引${ctx.forwardIndex}:==>${dayjs(res[0].max.Data.MsgSeq).format('YYYY-MM-DD HH:mm:ss')}`)
    }

    getConversationIdByMsg(msg:any)
    {
        let other = null
        
        if (msg.Data.FromUserName.string == msg.Wxid) {
            other = msg.Data.ToUserName.string
        }else{
            other =  msg.Data.FromUserName.string
        }

        return other
    }

    async pushMessageToMongo(msg: any) {

        let conversationID = this.getConversationIdByMsg(msg);
        let doc_id = `${msg.Wxid}_${conversationID}`;
        let ctx = await this.ensureConversationExist(doc_id);

        // await this.msgColl.updateOne({ doc_id: "si_2062181717_8171668571" },
        //     {
        //         $push: {
        //             msgs: msg as never
        //         },
        //     }, { upsert: true })


        // await this.msgColl.updateOne({ doc_id: `${msg.Wxid}_${conversationID}` },
        //     {
        //         $setOnInsert: {
        //             maxSeq: 0,
        //             msgs: []
        //         }
        //     }, 
        // { upsert: true })


        let update = { $set: {} }
        ctx.write(1, (index) => {
            update['$set'][`msgs.${index}`] = msg
            console.log(index)
            console.log(`存储到Mongo:${index}:${dayjs(msg.Data.MsgSeq).format('YYYY-MM-DD HH:mm:ss.SSS')}`)
        })

        await this.updateContext(doc_id,ctx, update)
    }
}